package org.example.process;

import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.TimerService;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.example.data.WaterSensor;
import org.example.function.WaterSensorMapFunction;

import java.time.Duration;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * 滑动时间窗口，实现TopN
 */
public class TopNDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<String> source = env.socketTextStream("localhost", 9999);
        SingleOutputStreamOperator<WaterSensor> waterSensorStream = source.map(new WaterSensorMapFunction());

        WatermarkStrategy<WaterSensor> waterSensorWatermarkStrategy = WatermarkStrategy
                //指定乱序的Watermark生成，延时。
                .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner((WaterSensor waterSensor, long l) -> {
                    //返回的时间戳 毫秒
                    return waterSensor.getTs() * 1000;
                });

        SingleOutputStreamOperator<WaterSensor> sensorDS = waterSensorStream.assignTimestampsAndWatermarks(waterSensorWatermarkStrategy);

        //思路一：所有数据放在一起，用Hashmap统一，key为水位
        //最近10s=窗口长度，每5s输出=滑动步长
        sensorDS.windowAll(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))
                .process(new MyTopNPAWF())
                .print();


        env.execute("PartitionDemo");
    }

    public static class MyTopNPAWF extends ProcessAllWindowFunction<WaterSensor, String, TimeWindow> {
        private static final long serialVersionUID = 1L;

        @Override
        public void process(ProcessAllWindowFunction<WaterSensor, String, TimeWindow>.Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {
            HashMap<Integer, Integer> vcCountMap = new HashMap<>();
            elements.forEach(waterSensor -> {
                vcCountMap.put(waterSensor.getVc(), vcCountMap.getOrDefault(waterSensor.getVc(), 0) + 1);
            });

            StringBuilder outStr = new StringBuilder();
            outStr.append("<===========================>\n");
            Integer i = 0;
            //降序取前两名
            List<Map.Entry<Integer, Integer>> toSort = new ArrayList<>();
            for (Map.Entry<Integer, Integer> o : vcCountMap.entrySet()) {
                toSort.add(o);
            }
            toSort.sort((o1, o2) -> o2.getValue() - o1.getValue());
            long limit = 2;
            for (Map.Entry<Integer, Integer> o : toSort) {
                if (limit-- == 0) break;
                i++;
                outStr.append("Top: " + i + " 的vc: " + o.getKey() + " 值为:" + o.getValue() + "窗口结束时间：" + DateFormatUtils.format(context.window().getEnd(), "yyyy-MM-dd HH:mm:ss") + "\n");
            }
            outStr.append("<===========================>\n");
            out.collect(outStr.toString());
        }
    }
}